/**
 * 
 */
package com.lorizz.picpub.task;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * @author wind hong
 * 
 */
public class TaskExecutionManager {

	private BlockingQueue<Long> taskQueue;
	private TaskExecutor<Long> executor;
	private ExecutorService executorService;
	private int threads;
	private String name;
	private Log log = LogFactory.getLog(getClass());

	public void setTaskQueue(BlockingQueue<Long> taskQueue) {
		this.taskQueue = taskQueue;
	}

	public void setExecutor(TaskExecutor<Long> executor) {
		this.executor = executor;
	}

	public void setExecutorService(ExecutorService executorService) {
		this.executorService = executorService;
	}

	public void setThreads(int threads) {
		this.threads = threads;
	}

	public void setName(String name) {
		this.name = name;
	}

	private Long getTask() throws InterruptedException {
		return taskQueue.take();
	}

	public void init() {
		log.debug(String.format("Initializing task execution manager %s ...", name));
		for (int i = 0; i < threads; i++) {
			executorService.execute(new TaskExecutionWorker(name + "[" + i + "]"));
		}
	}

	public void destroy() {
		executorService.shutdown();
	}

	class TaskExecutionWorker implements Runnable {
		private String name;

		public TaskExecutionWorker(String name) {
			this.name = name;
			log.debug(String.format("Initializing Worker[%s]", name));
		}

		@Override
		public void run() {
			while (true) {
				Long id;
				try {
					id = getTask();
				} catch (InterruptedException e) {
					log.error(String.format("Worker[%s] get task error", name), e);
					continue;
				}
				executor.execute(id);
			}
		}

	}

}
